package major;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.*;

public class Observed {
    public static void main(String[] args) {
//        Map<String, Object> config = new HashMap<>();
//        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.0.7:9092");
//        config.put(ProducerConfig.CLIENT_ID_CONFIG,"mei_mei");
//        config.put(ProducerConfig.ACKS_CONFIG,"all");
//        config.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,CountInter.class.getName());
//        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
//        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
//        Producer producer = new KafkaProducer<>(config);
        Producer<String, String> producer = new ProducerBuiler("mei_mei").build();
        List<String> topics = Arrays.asList("blue", "grey");
        Scanner scanner = new Scanner(System.in);
        while (true){
            String s = scanner.nextLine();
            if (s.equals("quit")) break;
            Integer age = 0;
            try {
                age = Integer.parseInt(s);
            }catch (Exception e){
                age=18;
            }
            for (String topic : topics) {
                producer.send(new ProducerRecord(topic, age + ""), new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        System.out.println(Thread.currentThread().getName()+" --- "+recordMetadata.topic());
                    }
                });
            }


        }
        producer.close();

    }
    static class ProducerBuiler{
        private Producer<String,String> producer;

        public ProducerBuiler(String producerName){
            Map<String, Object> config = new HashMap<>();
            config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.0.7:9092");
            config.put(ProducerConfig.CLIENT_ID_CONFIG,producerName);
            config.put(ProducerConfig.ACKS_CONFIG,"all");
            config.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,CountInter.class.getName());
            config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            this.producer = new KafkaProducer<>(config);

        }

        public Producer<String, String> build() {
            return producer;
        }
    }
}
